package com.yungu.swift.socket.communication.server.handler;


import com.google.common.collect.Lists;
import com.google.gson.Gson;
import com.yungu.swift.socket.communication.message.LoginMessage;
import com.yungu.swift.socket.communication.message.base.Message;
import com.yungu.swift.socket.communication.message.base.MessageType;
import com.yungu.swift.socket.communication.server.handler.strategy.MessageHandlerContext;
import com.yungu.swift.socket.component.ChannelKeeper;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Configuration;

import java.util.ArrayList;
import java.util.List;

import static com.yungu.swift.socket.communication.message.base.MessageType.*;

/**
 * webSocket handler
 *
 * @author cuixiuyin
 * @since 2018/1/6
 */
@Configuration
@ChannelHandler.Sharable
public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {

    private static final Logger logger = LoggerFactory.getLogger(WebSocketServerHandler.class);

    //握手
    private WebSocketServerHandshaker handShaker;

    /**
     * 读取完成
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    /**
     * 消息接受
     *
     * @param ctx     通道容器
     * @param message 报文
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object message) throws Exception {
        //普通HTTP接入
        if (message instanceof FullHttpRequest) {
            //强转
            FullHttpRequest request = (FullHttpRequest) message;
            boolean successDecode = request.decoderResult().isSuccess();
            String headerUpgrade = "";
            if (request.headers().contains("Upgrade")) {
                //消息头-包含Upgrade字段(协议升级)
                headerUpgrade = request.headers().get("Upgrade");
            }
            //如果http解码失败 则返回http异常 并且判断消息头有没有包含Upgrade字段(协议升级)
            if (!successDecode || (!"websocket".equals(headerUpgrade))) {
                sendHttpResponse(ctx, request, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
                return;
            }
            //执行协议升级 —— 构造握手响应返回
            WebSocketServerHandshakerFactory ws = new WebSocketServerHandshakerFactory("", null, false);
            handShaker = ws.newHandshaker(request);
            //版本不支持
            if (handShaker == null) {
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
            } else {
                handShaker.handshake(ctx.channel(), request);
            }

        }
        //websocket帧类型 已连接
        if (message instanceof WebSocketFrame) {
            handleWebSocketFrame(ctx, (WebSocketFrame) message);
        }
    }


    /**
     * webSocket帧
     *
     * @param ctx
     * @param webSocketFrame
     */
    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame webSocketFrame) {
        //判断是否关闭链路指令
        if (webSocketFrame instanceof CloseWebSocketFrame) {
            CloseWebSocketFrame closeWebSocketFrame = (CloseWebSocketFrame) webSocketFrame;
            handShaker.close(ctx.channel(), closeWebSocketFrame.retain());
            ctx.close();
            //移除
            ChannelKeeper.remove(ctx.channel());
            return;
        }
        //判断是否Ping消息 -- ping/pong心跳包
        if (webSocketFrame instanceof PingWebSocketFrame) {
            ctx.writeAndFlush(new PongWebSocketFrame(webSocketFrame.content().retain()));
            return;
        }
        //仅支持文本消息， 不支持二进制消息
        if (!(webSocketFrame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException(String.format("%s frame types not supported", webSocketFrame.getClass().getName()));
        }

        TextWebSocketFrame textWebSocketFrame = (TextWebSocketFrame) webSocketFrame;
        //处理消息
        dealMessage(ctx, textWebSocketFrame.text());
    }


    /**
     * 处理消息
     *
     * @param channelHandlerContext
     * @param text
     */
    private void dealMessage(ChannelHandlerContext channelHandlerContext, String text) {
        logger.info("received message:-->{}", text);
        //报文类型做判断
        Gson gson = new Gson();
        Message message = gson.fromJson(text, Message.class);
        // 1. 非登陆报文，需要判断客户端是否登陆
        // 2. todo 后台报文需要加权限校验，目前先放过
        ArrayList<MessageType> notAuthTypes = Lists.newArrayList(LOGIN, NEW_JOIN_ORDER_ADMIN_NOTIFY,
                NEW_JOIN_ORDER_CALL,ADMIN_USER_LOGIN);
        if (ChannelKeeper.get(message.getClientUuid()) == null && !notAuthTypes.contains(message.getType())) {
            //说明未登录，或者连接断了，服务器向客户端发起登录请求，让客户端重新登录
            LoginMessage loginMsg = new LoginMessage(LoginMessage.From.SERVER);
            channelHandlerContext.channel().writeAndFlush(new TextWebSocketFrame(gson.toJson(loginMsg)));
        } else {
            MessageHandlerContext.handler(channelHandlerContext, text, message.getType());
        }
        ReferenceCountUtil.release(message);
    }

    /**
     * sendHttpResponse 发送http响应
     *
     * @param ctx
     * @param request
     * @param response
     */
    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest request, FullHttpResponse response) {
        //返回给客户端
        boolean responseStatusOK = (response.status().code() == HttpResponseStatus.OK.code());
        if (!responseStatusOK) {
            ByteBuf buf = Unpooled.copiedBuffer(response.status().toString(), CharsetUtil.UTF_8);
            response.content().writeBytes(buf);
            buf.release();
            HttpUtil.setContentLength(response, response.content().readableBytes());
        }
        //如果不是keepAlive那么就关闭连接
        ChannelFuture f = ctx.channel().writeAndFlush(response);
        boolean isKeepAlive = HttpUtil.isKeepAlive(response);
        if (!isKeepAlive || !responseStatusOK) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }

    /**
     * 异常捕捉
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.info("exceptionCaught:cause-->" + cause);
        ctx.close();
        ChannelKeeper.remove(ctx.channel());
        logger.info("exceptionCaught over. channel-->" + ctx.channel());
        super.exceptionCaught(ctx, cause);
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {

        logger.info("客户端主动断开channel：{}", ctx.channel().id());
        //补充手动关闭通道缓存
        ChannelKeeper.remove(ctx.channel());

    }
}